package com.czl.myRabbitMq.mqServer.dataCenter;

import com.czl.myRabbitMq.commom.exception.mqException;
import com.czl.myRabbitMq.mqServer.core.Binding;
import com.czl.myRabbitMq.mqServer.core.Exchange;
import com.czl.myRabbitMq.mqServer.core.MSGQueue;
import com.czl.myRabbitMq.mqServer.core.Message;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 这个类来对内存中的数据进行统一的管理
 * 为此我们需要数据结构来对内存中的数据
 * 进行存储
 * 同时这个类很可能有很多个对象同时访问
 * 所以类中的对象需要使用线程安全的数据
 * 结构或者进项上锁操作
 */
@Slf4j
public class MemoryDataManager {
    //存储交换机的key表示ExchangeName Value表示Exchange
    private final ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //存储消息队列也是Key表示QueueName value表示Queue
    private final ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //存储绑定因为涉及到三个对象的关系，所以我们使用嵌套的哈表表来进行存储
    //第一个String表示的是ExchangeName 里面的value为一个嵌套hashmap 其中嵌套hashmap的第一个参数为
    //QueueName 第二个参数为绑定
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingMap =
            new ConcurrentHashMap<>();
    //同时我们要表示内存中总的的消息，就是所有交换机内的消息都会在这里进行展示 第一个参数为MessageId第二个参数为Message
    private final ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //这里是一个队列中的消息 第一个参数为队列名 第二个为队列中的消息
    private final ConcurrentHashMap<String, LinkedList<Message>> messageListMap = new ConcurrentHashMap<>();
    //我们为了实现另外一种确认应答的机制 就是程序获取了消息后需要返回一个请求手动来删除ID这里我们还需要引入一个
    //map来解决 第一个参数表示当前队列，后面嵌套中的第一个参数表示取走了但是还未返回ack的MessageId 后面为Message
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> waitAckMessageMap =
            new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(), exchange);
        log.info("插入交换机成功" + exchange.getName());
    }
    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        log.info("删除交换机成功" + exchangeName);
    }
    public Exchange selectExchange(String exchangeName){
        log.info("查询交换机" + exchangeName);
        return exchangeMap.get(exchangeName);

    }
    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(), queue);
        log.info("插入队列成功" + queue.getName());
    }
    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        log.info("删除队列成功" + queueName);
    }
    public MSGQueue selectQueue(String queueName){
        log.info("查询交换机");
        return queueMap.get(queueName);
    }

    /**
     * 因为 BindingMap 是一个嵌套的哈希表所以在进行插入的时候
     * 我们应该查询Binding中是否存在QueueMap没有的话就需要new
     * 一个出来
     * @param binding
     */
    @SneakyThrows
    public void insertBinding(Binding binding){
//        ConcurrentHashMap<String,Binding> map = bindingMap.get(binding.getExchangeName());
//        if(map == null){
//            map = new ConcurrentHashMap<>();
//            map.put(binding.getQueueName(),binding);
//        }
        //这些代码表示先查询是够存在绑定中的队列，要是不存在的话 就需要new出来
        //为了简化代码，我们可以将代码简化为一下代码 这行代码其中的computeIfAbsent就是map的get方法
        //只不过要是不存在相应的值，就会之后后面的语句 这里就是要是不存在的话 就进行一个创建 并且将新
        //创建的内容插入到原来的map之中
        ConcurrentHashMap<String, Binding> map = bindingMap.computeIfAbsent(binding.getExchangeName(),
                t -> new ConcurrentHashMap<>());

        synchronized (map) {
            //先查询绑定是否已经存在，要是不存在才进行添加
            if (map.get(binding.getQueueName()) != null)
                throw new mqException("Binding已经存在" + binding.getBindingKey());

            map.put(binding.getQueueName(), binding);
            log.info("创建绑定成功");
        }
    }
    //这里的查找binding写2个方法
    //根据交换机和队列的名称进行查找
    @SneakyThrows
    public Binding selectBinding(String ExchangeName, String queueName){
        ConcurrentHashMap<String, Binding> map = bindingMap.get(ExchangeName);
        if(map == null) {
            log.info("所要查找的交换机不存在对应的绑定队列");
            return null;
        }
        log.info("#{}", map.get(queueName));
        return map.get(queueName);
    }

    //查找一个交换机下所有的綁定
    public ConcurrentHashMap<String, Binding> selectBindings(String ExchangeName){
        return bindingMap.get(ExchangeName);
    }
    //删除绑定
    @SneakyThrows
    public void deleteBinding(Binding binding){
        ConcurrentHashMap<String, Binding> map = bindingMap.get(binding.getExchangeName());
        if(map == null)
            throw new mqException("该交换机下不存在队列" + binding.getExchangeName());
        map.remove(binding.getQueueName());
        log.info("删除绑定成功" + binding.getExchangeName() + "and" + binding.getQueueName());
    }
    public Message selectMessage(String messageId){
        return messageMap.get(messageId);
    }
    public void deleteMessage(String messageId){
        messageMap.remove(messageId);
        log.info("删除消息成功" +messageId);
    }
    //往中的消息中心插入一条数据
    public void insertMessage(Message message){
        messageMap.put(message.getMessageId(), message);
        log.info("插入消息成功" + message.getMessageId());
    }

    //发送消息到指定的队列
    public void sendMessage(MSGQueue queue, Message message){
        //先获取到指定的队列  同时这个computeIfAbsent本身就是线程安全的
        //这个方法看75行注释
        LinkedList<Message> messages = messageListMap.computeIfAbsent(queue.getName(),
                t -> new LinkedList<>());
        synchronized (messages){
            messages.add(message);
        }
        //往指定队列插入消息之后，将该消息插入到中的消息中心里
        insertMessage(message);
        log.info("往" + queue.getName() + "写入消息" + message.getMessageId());

    }

    //从队列中获取一条消息,类似于队列一样,先存储的消息先进行获取

    public Message pollMessage(String queueName){
        //先获取到指定消息的连标记结构
        LinkedList<Message> messages = messageListMap.get(queueName);

        if(messages == null || messages.size() == 0)
            return null;
        synchronized (messages){
            //对链表进行头删 就等于取走第一个元素
            Message message = messages.remove(0);
            log.info("从队列中取走元素" + message.getMessageId());
            return message;
        }


    }

    //查询队列中剩下的消息条数
    public int getMessageNum(String queueName){
        LinkedList<Message> messages = messageListMap.get(queueName);
        if(messages == null)
            return 0;
        return messages.size();
    }

    //添加一个未确认的消息
    public void addMessageWaitACK(MSGQueue queue, Message message){
        //因为这里的2个方法都是线程安全的 所以不需要上锁
        ConcurrentHashMap<String, Message> messages = waitAckMessageMap.computeIfAbsent(queue.getName(),
                t -> new ConcurrentHashMap<>());
        messages.put(message.getMessageId(), message);
        log.info("添加未确认消息成功" + message.getMessageId());
    }
    //删除一个未确认的消息
    public void deleteMessageWaitACK(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messages = waitAckMessageMap.get(queueName);
        if(messages == null)
            return;
        messages.remove(messageId);
        log.info("删除未确认的消息成功" + messageId);
    }
    //查找一个指定的未确认的消息
    public Message selectMessageWaitACK(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messages = waitAckMessageMap.get(queueName);
        if(messages == null){
            log.info("队列" + queueName + "中不存在未确认的消息");
            return null;
        }
        return messages.get(messageId);
    }

    //启动服务器的时候从硬盘读取数据恢复到内存中
    //并且恢复顺序为
    // 1. 恢复交换机数据
    // 2. 恢复队列数据
    // 3. 恢复绑定顺序
    // 4. 恢复队列中消息顺序
    public void recoverData(DiskDataManager diskdataManager){
        //先将管理内存数据的数据结构内容全部清空防止特殊情况
        exchangeMap.clear();
        queueMap.clear();
        messageMap.clear();
        messageListMap.clear();
        bindingMap.clear();
        //先进行交换机的恢复
        List<Exchange> exchanges = diskdataManager.selectAllExchange();
        for(Exchange exchange : exchanges){
            exchangeMap.put(exchange.getName(), exchange);
        }

        //然后进行队列数据和消息数据的恢复
        List<MSGQueue> queues = diskdataManager.selectAllQueue();
        for(MSGQueue queue : queues){
            //恢复队列的数据
            queueMap.put(queue.getName(), queue);
            //查询出当前队列存在那些消息
            LinkedList<Message> messageList = diskdataManager.loudAllMessage(queue.getName());
            //恢复队列中的消息数据
            messageListMap.put(queue.getName(),messageList);
            //恢复队列中消息数据后，需要将消息插入总消息中， 也就是消息中心
            for(Message message : messageList){
                messageMap.put(message.getMessageId(), message);
            }
        }

        //恢复绑定中的数据
        List<Binding> bindings = diskdataManager.selectAllBinding();
        //因为绑定是嵌套循环
        for(Binding binding : bindings){
            ConcurrentHashMap<String,Binding> map = bindingMap.computeIfAbsent(binding.getExchangeName(),
                    t -> new ConcurrentHashMap<>());
            map.put(binding.getQueueName(), binding);
        }

    }
}
